package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/31
 *
 *  不重要。
 *          flink主要做流处理。 DataSetApi
 *          批处理是历史遗产
 *
 *
 *  ---------------
 *      套路
 *          ①创建运行环境
 *          ②从运行环境中读数据，封装为数据模型
 *          ③调用api计算
 *          ④输出
 */
public class Demo1_BatchDemo
{
    public static void main(String[] args) throws Exception {

        //①创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //②从运行环境中读数据，封装为数据模型
        DataSource<String> ds = env.readTextFile("data/words.txt");
        //③调用api计算  输入: 一行  输出: (单词,1)
        ds
            .flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>()
            {
                /*
                        Collector<T> out:  输出结果的收集器。
                            调用收集器把想输出的数据输出
                 */
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {

                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word,1));
                    }
                }
            })
            /*
                 当前的数据类型是 Tuple类型，调用  groupBy(int fields), 按照tuple中字段的位置分组。位置索引从0开始！

                 当前的数据类型是 POJO(Bean)类型，调用  groupBy(String fields)， 按照POJO中字段名分组

             */
            .groupBy(0)
            .sum(1)
            //④输出
            .print();

    }
}
